package com.hudson.admin.client;


import com.google.gson.reflect.TypeToken;
import com.hudson.admin.client.handler.ReconnectionHandler;
import com.hudson.admin.config.NameSrvConfig;
import com.hudson.codec.MessageCodecFactory;
import com.hudson.codec.domain.RemotingCommand;
import com.hudson.common.enums.MessageTypeEnum;
import com.hudson.core.json.JSONUtil;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.handler.timeout.IdleStateHandler;
import jakarta.annotation.Resource;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.stereotype.Component;

import java.net.InetSocketAddress;
import java.nio.charset.StandardCharsets;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;

/**
 * @Author: 渔小鲤
 * @DateTime: 2025/7/24 21:59
 **/
@Component
public class NameSrvClient implements InitializingBean {
    private static final Logger logger = LoggerFactory.getLogger(NameSrvClient.class);

    @Resource
    private NameSrvConfig config;
    private static Bootstrap bootstrap;

    private static Channel nameSrvClient;

    private static CompletableFuture<Map<String, String>> result;

    @Override
    public void afterPropertiesSet() throws Exception {
        bootstrap = new Bootstrap()
                .group(new NioEventLoopGroup(4))
                .channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<NioSocketChannel>() {
                    @Override
                    protected void initChannel(NioSocketChannel sc) throws Exception {
                        sc.pipeline().addLast(new LengthFieldBasedFrameDecoder(1024 * 1024, 8, 4, 0, 0));
                        sc.pipeline().addLast(MessageCodecFactory.getCodec(config.getSerializableMethod()));
                        sc.pipeline().addLast(new IdleStateHandler(0, 10, 0, TimeUnit.SECONDS)); // 写空闲4秒触发 WRITE_IDLE
                        sc.pipeline().addLast(new ChannelInboundHandlerAdapter() {
                            @Override
                            public void userEventTriggered(ChannelHandlerContext ctx, Object evt) {
                                if (evt instanceof IdleStateEvent) {
                                    IdleStateEvent event = (IdleStateEvent) evt;
                                    if (event.state() == IdleState.WRITER_IDLE) {
                                        // 发送心跳包
                                        logger.debug("produce 空闲，发送心跳至 broker...");
                                        ctx.writeAndFlush(RemotingCommand.builder().type(MessageTypeEnum.IDLE_MESSAGE.getType()).build()); // 或者发送自定义心跳消息
                                    }
                                }
                            }
                        });
                        sc.pipeline().addLast(new ReconnectionHandler(config.getAddress(), config.getPort()));
                        sc.pipeline().addLast(new SimpleChannelInboundHandler<RemotingCommand>() {
                            @Override
                            protected void channelRead0(ChannelHandlerContext channelHandlerContext, RemotingCommand message) throws Exception {
                                if (MessageTypeEnum.GET_BROKER_INFO.equals(MessageTypeEnum.fromType(message.getType()))) {
                                    Map<String, String> resultMap = JSONUtil.fromJson(new String(message.getBody(), StandardCharsets.UTF_8), new TypeToken<Map<String, String>>() {
                                    });
                                    result.complete(resultMap);
                                }
                            }
                        });
                    }
                });

        nameSrvClient = bootstrap.connect(new InetSocketAddress(config.getAddress(), config.getPort())).sync().addListener(f -> {
            if (f.isSuccess()) {
                logger.debug("Admin 连接 NameSrv 成功");
            } else {
                System.exit(0);
            }
        }).channel();
    }


    public static ChannelFuture getChannel(String address, int port) {
        try {
            ChannelFuture channelFuture = bootstrap.connect(address, port);
            channelFuture.addListener((ChannelFutureListener) f -> {
                if (f.isSuccess()) {
                    logger.info("连接建立成功: {}:{}", address, port);
                } else {
                    logger.error("连接失败: {}:{}, 原因: {}", address, port, f.cause().getMessage());
                }
            });
            // 返回新创建的Channel，不阻塞等待
            return channelFuture;
        } catch (Exception e) {
            logger.error("创建连接时发生异常: {}:{}", address + ":" + port, e.getMessage());
            throw new RuntimeException(e);
        }
    }

    public Map<String, String> getBrokerInfo() {
        nameSrvClient.writeAndFlush(RemotingCommand.builder().type(MessageTypeEnum.GET_BROKER_INFO.getType()).build());
        result = new CompletableFuture<>();


        try {
            return result.get(3, TimeUnit.SECONDS);
        } catch (Exception e) {
            throw new RuntimeException("发送失败进行重试");
        }finally {
            try {
                BrokerClientManege.connect(result.get());
            } catch (Exception e) {
                logger.error("连接broker失败");
            }
        }
    }


    public static void setNameSrvChanel(Channel newChannel) {
        nameSrvClient = newChannel;
    }
}
